package cn.qdhlm.frontendprocessor.core.callback;


import cn.qdhlm.frontendprocessor.common.config.MqttConfig;
import cn.qdhlm.frontendprocessor.common.helper.RocketMqProducer;
import lombok.extern.log4j.Log4j2;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import io.netty.buffer.ByteBufUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


/**
 * @Name:MqttCallback
 * @description:
 * @author: xinY
 **/
@Log4j2
@Component
public class MqttReceiveCallback implements MqttCallback {
    @Autowired
    private RocketMqProducer producer;

    @Override
    public void connectionLost(Throwable throwable) {
        log.error("mqtt connection lost", throwable);
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {

        // 从mqtt服务器get data
//        String hexStr = ByteBufUtil.hexDump(Unpooled.wrappedBuffer(mqttMessage.getPayload()));
        String hexStr = ByteBufUtil.hexDump(mqttMessage.getPayload());
        log.info("arrived: messageId[{}], topic[{}], qos[{}], hexStr[{}]", mqttMessage.getId(), topic, mqttMessage.getQos(), hexStr);

        // 把mqtt的消息发送到rocketmq 服务器上
        if (MqttConfig.topics[0].equals(topic)
                || MqttConfig.topics[1].equals(topic)
                || MqttConfig.topics[2].equals(topic)
                || MqttConfig.topics[3].equals(topic)) {
            producer.send("DeviceData", "mqtt", topic, mqttMessage.getPayload());
        }

        // nb-mqtt的消息发送rocketmq 服务器上
        else if (MqttConfig.topics[4].equals(topic) || MqttConfig.topics[5].equals(topic)) {
            producer.send("DeviceData", "nb-mqtt", topic, mqttMessage.getPayload());
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        try {
            log.info("delivery complete: messageId[{}] => {}", iMqttDeliveryToken.getMessage().getId(), iMqttDeliveryToken.isComplete());
        } catch (MqttException e) {
            log.error("delivery complete exception: ", e);
        }
    }
}